package com.ytktt.study.week9.task3.rpcfx.registry;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.ytktt.study.week9.task3.rpcfx.api.ServiceProviderDesc;
import com.ytktt.study.week9.task3.rpcfx.constant.RpcfxConstant;
import com.ytktt.study.week9.task3.rpcfx.enums.ErrResponseEnum;
import com.ytktt.study.week9.task3.rpcfx.exception.RpcfxException;
import com.ytktt.study.week9.task3.rpcfx.lb.LoadBalancer;
import com.ytktt.study.week9.task3.rpcfx.properties.RegistryProperties;
import com.ytktt.study.week9.task3.rpcfx.router.Router;
import com.ytktt.study.week9.task3.rpcfx.utils.StringFormatUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
import org.springframework.beans.factory.DisposableBean;

import java.util.List;
import java.util.Map;

/**
 * zk注册中心
 *
 * @author ytk
 */
public class ZookeeperRegistry implements Registry, DisposableBean {

    private final CuratorFramework client;
    private final Router router;
    private final LoadBalancer loadBalancer;
    private final Map<String, List<String>> serviceMap = Maps.newHashMap();

    public ZookeeperRegistry(RegistryProperties registryProperties, Router router, LoadBalancer loadBalancer) {
        this.router = router;
        this.loadBalancer = loadBalancer;
        // start zk client
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(registryProperties.getRetry().getBaseSleepTimeMs(), registryProperties.getRetry().getTimes());
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(registryProperties.getUrl())
                .namespace(RpcfxConstant.NAMESPACE)
                .retryPolicy(retryPolicy)
                .build();
        this.client = client;
        client.start();
    }

    @Override
    public String getServiceUrl(String serviceName, String group, String version) throws Exception {

        List<String> invokers = this.serviceMap.get(serviceName + ":" + group + ":" + version);
        // 1. 简单：从zk拿到服务提供的列表
        if (invokers == null) {
            invokers = this.getServiceProviderUrls(serviceName, group, version);
        }
        // 2. 挑战：监听zk的临时节点，根据时间更新这个List（主要，需要做个全局map保持每个服务的提供者List）
        List<String> urls = this.router.route(invokers);
        return this.loadBalancer.select(urls);
    }

    private List<String> getServiceProviderUrls(String serviceName, String group, String version) throws Exception {
        List<String> invokers = Lists.newArrayList();
        CuratorFrameworkState state = this.client.getState();
        if (CuratorFrameworkState.STOPPED.equals(state)) {
            throw new RpcfxException(ErrResponseEnum.REGISTRY_ACCESS_FAIL);
        }
        List<String> strings = this.client.getChildren().usingWatcher((Watcher) watchedEvent -> {
            try {
                getServiceProviderUrls(serviceName, group, version);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).forPath(StringFormatUtils.genBaseServicePath(serviceName, group, version));
        for (String s : strings) {
            invokers.add(RpcfxConstant.SCHEMA + s);
            System.out.println("provider: " + s);
        }
        this.serviceMap.put(StringFormatUtils.genRealServiceName(serviceName, group, version), invokers);
        return invokers;
    }

    @Override
    public void registryService(ServiceProviderDesc serviceProviderDesc) throws Exception {
        // String userServiceSescJson = JSON.toJSONString(userServiceSesc);

        String basePath = null;
        try {
            String groupPath = '/' + serviceProviderDesc.getGroup();
            if (null == this.client.checkExists().forPath(groupPath)) {
                this.client.create().withMode(CreateMode.PERSISTENT).forPath(groupPath, "group".getBytes());
            }
            String servicePath = groupPath + '/' + serviceProviderDesc.getServiceName();
            if (null == this.client.checkExists().forPath(servicePath)) {
                this.client.create().withMode(CreateMode.PERSISTENT).forPath(servicePath, "service".getBytes());
            }
            basePath = servicePath + '/' + serviceProviderDesc.getVersion();
            if (null == this.client.checkExists().forPath(basePath)) {
                this.client.create().withMode(CreateMode.PERSISTENT).forPath(basePath, "version".getBytes());
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }

        this.client.create().withMode(CreateMode.EPHEMERAL).
                forPath(basePath + '/' + serviceProviderDesc.getHost() + ':' + serviceProviderDesc.getPort(), "provider".getBytes());
    }

    @Override
    public void destroy() throws Exception {
        this.client.close();
    }
}
